package 实训二;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


//todo 新老用户
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //初始化环境和设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topic = "topic_log";
        String groupId = "base_log_app_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        source.print();

        //数据清洗转换结构  定义一个侧输出流的标签   用于标识脏数据流
        OutputTag<String> dirtyStream = new OutputTag<String>("dirtyStream") {};
        SingleOutputStreamOperator<String> filterStream = source.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                try {
                    collector.collect(s);
                } catch (Exception e) {
                    context.output(dirtyStream, s);
                }
            }
        });
        filterStream.print();

        // 把字符串转成JSONObject
        SingleOutputStreamOperator<JSONObject> mappedStream = filterStream.map(JSON::parseObject);

        // 按照mid字段进行分组  分组后所有属于同一个mid的记录会发往同一个算子实例处理
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(s -> s.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> firstValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                firstValueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDt", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");

                // 获取当前状态：首次访问日期
                String firstViewDt = firstValueState.value();
                Long ts = jsonObject.getLong("ts");
                String dt = DateFormatUtil.toDate(ts);

                //认为是新用户  状态中还没有记录首次访问时间
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstValueState.update(dt);
                    } else {
                        if (!firstViewDt.equals(dt)) {
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (firstViewDt == null) {
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstValueState.update(yesterday);
                    }
                }
                collector.collect(jsonObject);
            }
        });

        // 分流 page   start   err
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};

        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(new ProcessFunction<JSONObject, String>() {

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                // 提取err字段  如果存在则说明是一条错误日志
                JSONObject error = jsonObject.getJSONObject("err");
                if (error != null) {
                    // 输出到errTag  将整条原始日志(JSON字符串)发送到错误流
                    context.output(errTag, jsonObject.toJSONString());
                }
                jsonObject.remove("err");

                // 收集启动数据
                JSONObject start = jsonObject.getJSONObject("start");
                if (start != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    // 提取公共字段
                    JSONObject page = jsonObject.getJSONObject("page");
                    JSONObject common = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");

                    // 收集曝光数据  [{}，{}，{}]displays是一个数组 每一个元素是一个曝光事件
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null) {
                        // 遍历每个曝光事件，构建新的JSON对象
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("page", page);
                            displayObj.put("common", common);
                            displayObj.put("ts", ts);
                            context.output(displayTag, displayObj.toJSONString());
                        }
                    }
                    // 收集动作数据
                    JSONArray actions = jsonObject.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("page", page);
                            actionObj.put("common", common);
                            actionObj.put("ts", ts);
                            context.output(actionTag, actionObj.toJSONString());
                        }
                        // 手机页面数据
                        jsonObject.remove("displays");
                        jsonObject.remove("actions");

                        collector.collect(jsonObject.toJSONString());
                    }
                }
            }
        });

        separatedStream.print("page>>>>>>>>>>>>>>>>");
        separatedStream.getSideOutput(startTag).print("start>>>>>>>>>>>>>>>>>>");
        separatedStream.getSideOutput(displayTag).print("display>>>>>>>>>>>>>>>>>>>>");
        separatedStream.getSideOutput(actionTag).print("action>>>>>>>>>>>>>>>>>>>>>>");
        separatedStream.getSideOutput(errTag).print("err>>>>>>>>>>>>>>>>>>>>>>>>");

        //获取侧输出流并写入kafka
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errTag);

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String err_topic = "dwd_traffic_err_log";

        separatedStream.print("page>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(err_topic));

        env.execute();
    }
}
